MLOps e Produção

1 O que é MLOps?

A ponte entre protótipo e produção — por que 87% dos projetos de IA nunca chegam lá.

Por que projetos de IA falham em produção?

Os 6 pilares do MLOps

2 Observabilidade — rastreando tudo

Se você não pode ver, não pode debugar. Logging detalhado de cada etapa do pipeline.

🎮 Dashboard de observabilidade em tempo real

Veja métricas do sistema RAG da Nimbus Cloud sendo atualizadas em tempo real.

Ferramentas de observabilidade

Exemplo: Tracing com Langfuse

from langfuse import Langfuse
import time

langfuse = Langfuse(
    public_key="pk-...",
    secret_key="sk-...",
    host="https://cloud.langfuse.com"
)

def rag_pipeline(question):
    # Cria trace principal
    trace = langfuse.trace(
        name="rag-query",
        input=question,
        metadata={"user_id": "user_123"}
    )
    
    # 1. Embedding da pergunta
    with langfuse.span(name="embed-question", trace_id=trace.id) as span:
        start = time.time()
        q_embedding = embed(question)
        span.end(output={"dim": len(q_embedding), "latency_ms": (time.time()-start)*1000})
    
    # 2. Busca vetorial
    with langfuse.span(name="vector-search", trace_id=trace.id) as span:
        start = time.time()
        chunks = vector_db.search(q_embedding, top_k=5)
        span.end(output={"chunks": len(chunks), "latency_ms": (time.time()-start)*1000})
    
    # 3. Reranking
    with langfuse.span(name="rerank", trace_id=trace.id) as span:
        start = time.time()
        chunks = rerank(question, chunks, top_k=3)
        span.end(output={"reranked": len(chunks), "latency_ms": (time.time()-start)*1000})
    
    # 4. Geração pelo LLM
    with langfuse.span(name="llm-generation", trace_id=trace.id) as span:
        start = time.time()
        prompt = build_prompt(chunks, question)
        response = ollama.chat(model="llama3.1:8b", messages=[...])
        span.end(
            output=response,
            metadata={"tokens": response.usage.total_tokens, "latency_ms": (time.time()-start)*1000}
        )
    
    # 5. Avaliação automática
    faithfulness = evaluate_faithfulness(response, chunks)
    trace.score(name="faithfulness", value=faithfulness)
    
    trace.end(output=response)
    return response

# Cada requisição gera um trace completo no Langfuse
# Você pode ver: input, chunks recuperados, prompt, output, latências, scores

3 Versionamento — saber o que está em produção

Versionar modelos, datasets, prompts e código. Rollback em 1 clique.

O que versionar?

Ferramentas de versionamento

Exemplo: Versionamento com MLflow

import mlflow

mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("nimbus-rag")

with mlflow.start_run(run_name="v2.3-reranker-bge"):
    # Log de parâmetros
    mlflow.log_param("embedding_model", "nomic-embed-text")
    mlflow.log_param("reranker", "bge-reranker-v2-m3")
    mlflow.log_param("llm", "llama3.1:8b")
    mlflow.log_param("chunk_size", 400)
    mlflow.log_param("top_k", 5)
    
    # Log de métricas
    mlflow.log_metric("faithfulness", 0.92)
    mlflow.log_metric("answer_relevancy", 0.88)
    mlflow.log_metric("context_precision", 0.85)
    mlflow.log_metric("latency_p95", 2.3)
    
    # Log do modelo
    mlflow.sklearn.log_model(reranker, "reranker")
    
    # Log do prompt template
    with open("prompt_template.txt") as f:
        mlflow.log_text(f.read(), "prompt_template.txt")
    
    # Log do dataset de avaliação
    mlflow.log_artifact("golden_dataset_v2.json")

# Registrar modelo como "production"
mlflow.register_model("runs:/abc123/reranker", "nimbus-reranker")

# Agora você pode:
# - Ver histórico de experimentos
# - Comparar métricas entre versões
# - Fazer rollback para versão anterior
# - Saber exatamente o que está em produção

Prompts como código

# Trate prompts como código — versionados no Git

# prompts/rag_template_v2.txt
SYSTEM_PROMPT = """Você é o Nimbus Assistant.
REGRAS:
1. Responda APENAS com base no CONTEXTO.
2. Cite fontes no formato [Fonte: XXX].
3. Se não souber, diga "Não encontrei".

CONTEXTO:
{context}

PERGUNTA: {question}"""

# No código Python:
with open("prompts/rag_template_v2.txt") as f:
    template = f.read()

# Mudanças no prompt vão para o Git:
# git add prompts/
# git commit -m "feat: add citation requirement to RAG prompt"
# git push

# Se algo quebrar:
# git revert HEAD
# → prompt volta para versão anterior em segundos

4 Detecção de Drift — quando o modelo degrada

Modelos não são "treine uma vez, esqueça para sempre". Eles degradam com o tempo.

🎮 Simulador: Monitoramento de drift em tempo real

Veja a métrica de faithfulness do RAG ao longo do tempo. Quando cai abaixo do threshold, um alerta é disparado.

Tipos de drift

Como detectar drift automaticamente

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy
import pandas as pd

# Golden dataset: 100 perguntas com respostas ideais
golden_dataset = pd.read_json("golden_dataset_v2.json")

def check_drift():
    # Roda avaliação no dataset
    results = evaluate(
        dataset=golden_dataset,
        metrics=[faithfulness, answer_relevancy]
    )
    
    current_faithfulness = results["faithfulness"]
    current_relevancy = results["answer_relevancy"]
    
    # Compara com baseline (versão anterior)
    baseline = load_baseline()  # do MLflow
    
    # Calcula degradação
    faithfulness_drop = baseline.faithfulness - current_faithfulness
    relevancy_drop = baseline.relevancy - current_relevancy
    
    # Alertas
    if faithfulness_drop > 0.05:
        send_alert(f"⚠️ Faithfulness caiu {faithfulness_drop:.2f}!")
    
    if relevancy_drop > 0.05:
        send_alert(f"⚠️ Answer relevancy caiu {relevancy_drop:.2f}!")
    
    # Log no MLflow
    mlflow.log_metric("faithfulness", current_faithfulness)
    mlflow.log_metric("answer_relevancy", current_relevancy)
    
    return results

# Roda diariamente (cron job)
# 0 2 * * * python check_drift.py

O que fazer quando detectar drift?

5 Avaliação contínua — RAGAS em produção

Provar que seu RAG funciona — não só no demo, mas todo dia.

Métricas do RAGAS

Golden dataset — sua "prova" do RAG

# golden_dataset.json
[
  {
    "question": "Qual o timeout padrão do endpoint /v2/sync?",
    "ground_truth": "O timeout padrão é 30 segundos.",
    "contexts": ["Docs API v3.2: timeout default 30s"],
    "category": "technical"
  },
  {
    "question": "Quais NFs de equipamentos chegaram em junho?",
    "ground_truth": "NF 12345 da Tech Ltda, R$ 48.500, recebida em 03/06.",
    "contexts": ["NF 12345 | Tech Ltda | 03/06/2025 | R$48.500"],
    "category": "financial"
  },
  {
    "question": "Como aumentar o timeout?",
    "ground_truth": "Edite config.yaml, adicione NIMBUS_TIMEOUT: 120 e reinicie.",
    "contexts": ["Runbook RB-018: editar config.yaml..."],
    "category": "how-to"
  }
  // ... 100+ exemplos cobrindo todas as categorias
]

# Como construir o golden dataset:
# 1. Colete perguntas reais de usuários
# 2. Respostas ideais escritas por especialistas
# 3. Contextos relevantes (chunks que deveriam ser recuperados)
# 4. Categorize por tipo (técnica, financeira, como-fazer, etc.)
# 5. Mantenha atualizado — adicione novos casos regularmente

🎮 Simulador: Avaliação RAGAS

A/B testing de prompts

# Compare dois prompts em produção
import random

def rag_with_ab_test(question):
    # 50% prompt A, 50% prompt B
    variant = "A" if random.random() > 0.5 else "B"
    
    if variant == "A":
        prompt = PROMPT_V1  # versão atual
    else:
        prompt = PROMPT_V2  # nova versão
    
    response = generate(prompt, question)
    
    # Log no Langfuse com variant
    langfuse.trace(
        name="rag-query",
        metadata={"variant": variant},
        output=response
    )
    
    # Depois de 1000 requisições, compare métricas
    # Langfuse mostra: faithfulness A vs B, latência A vs B, etc.
    
    return response

# Após 1 semana:
# - Variant A: faithfulness 0.89, latência 1.8s
# - Variant B: faithfulness 0.93, latência 1.9s
# → Promove B para 100% do tráfego

6 CI/CD para IA — automatizando tudo

Do commit ao deploy em produção — sem intervenção manual.

🎮 Pipeline CI/CD interativo

Veja cada etapa do pipeline sendo executada automaticamente após um commit.

Exemplo: GitHub Actions para RAG

# .github/workflows/rag-ci.yml
name: RAG CI/CD

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install ragas pytest
      
      - name: Run unit tests
        run: pytest tests/
      
      - name: Run RAG evaluation
        run: |
          python evaluate_rag.py \
            --dataset golden_dataset.json \
            --output results.json
      
      - name: Check metrics threshold
        run: |
          python check_metrics.py \
            --results results.json \
            --min-faithfulness 0.90 \
            --min-relevancy 0.85
      
      - name: Upload to MLflow
        if: success()
        run: |
          python log_to_mlflow.py \
            --run-name "ci-${{ github.sha }}" \
            --results results.json
  
  deploy:
    needs: test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    steps:
      - name: Deploy to production
        run: |
          curl -X POST ${{ secrets.DEPLOY_WEBHOOK }} \
            -H "Authorization: Bearer ${{ secrets.DEPLOY_TOKEN }}"
      
      - name: Smoke test
        run: |
          python smoke_test.py \
            --endpoint ${{ secrets.PROD_URL }} \
            --question "Qual o timeout do /v2/sync?"

Etapas do pipeline

7 Custos e otimização

LLMs cobram por token. Sem otimização, a conta chega inesperada.

🎮 Calculadora de custos

Estime o custo mensal do seu sistema RAG baseado no uso.

Estratégias de otimização

Exemplo: Semantic Cache

import chromadb
from sklearn.metrics.pairwise import cosine_similarity

class SemanticCache:
    def __init__(self, threshold=0.95):
        self.db = chromadb.Client()
        self.cache = self.db.get_or_create_collection("semantic_cache")
        self.threshold = threshold
    
    def get(self, question):
        # Busca pergunta semanticamente similar no cache
        q_embedding = embed(question)
        results = self.cache.query(
            query_embeddings=[q_embedding],
            n_results=1
        )
        
        if results["distances"][0][0] > self.threshold:
            return results["documents"][0][0]  # Cache hit!
        return None  # Cache miss
    
    def set(self, question, answer):
        # Adiciona ao cache
        q_embedding = embed(question)
        self.cache.add(
            embeddings=[q_embedding],
            documents=[answer],
            ids=[hash(question)]
        )

# Uso no pipeline RAG
cache = SemanticCache(threshold=0.95)

def rag_with_cache(question):
    # 1. Tenta cache
    cached = cache.get(question)
    if cached:
        log_metric("cache_hit", 1)
        return cached  # Economia de $$$!
    
    # 2. Cache miss — roda pipeline normal
    log_metric("cache_hit", 0)
    response = rag_pipeline(question)
    
    # 3. Salva no cache
    cache.set(question, response)
    
    return response

# Com 30% de cache hit rate:
# - 1500 req/dia × 30% = 450 req economizadas
# - Economia: ~$15/dia = $450/mês

Comparação de custos

8 Escalabilidade e arquitetura

De 10 para 10.000 requisições por minuto — sem quebrar.

Arquitetura de produção

# Arquitetura escalável para RAG

# 1. API Gateway (load balancer)
#    - Rate limiting
#    - Autenticação
#    - Roteamento

# 2. Retrieval Service (múltiplas instâncias)
#    - Embedding da pergunta
#    - Busca vetorial (ChromaDB/Qdrant)
#    - Reranking
#    - Horizontal scaling: Kubernetes replicas

# 3. Generation Service (múltiplas instâncias)
#    - Montagem de prompt
#    - Chamada ao LLM
#    - Streaming de resposta
#    - GPU nodes: A100/H100 para LLMs grandes

# 4. Cache Layer (Redis)
#    - Semantic cache
#    - Cache de respostas frequentes
#    - TTL configurável

# 5. Vector DB (cluster)
#    - ChromaDB cluster ou Qdrant cloud
#    - Sharding por coleção
#    - Replicas para leitura

# 6. Observability Stack
#    - Langfuse (tracing)
#    - Prometheus + Grafana (métricas)
#    - Loki (logs)

🎮 Simulador: Escalabilidade

Checklist de produção

Docker Compose para produção

# docker-compose.prod.yml
version: '3.8'
services:
  api:
    build: .
    ports: ["8080:8080"]
    environment:
      - REDIS_URL=redis://redis:6379
      - CHROMA_URL=http://chromadb:8000
      - OLLAMA_URL=http://ollama:11434
      - LANGFUSE_HOST=http://langfuse:3000
    depends_on: [redis, chromadb, ollama, langfuse]
    deploy:
      replicas: 3  # 3 instâncias para load balancing
      resources:
        limits:
          cpus: '2'
          memory: 4G
  
  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
    volumes: ["redis_data:/data"]
    command: redis-server --appendonly yes
  
  chromadb:
    image: chromadb/chroma:latest
    ports: ["8000:8000"]
    volumes: ["chroma_data:/chroma/chroma"]
    environment:
      - ANONYMIZED_TELEMETRY=False
  
  ollama:
    image: ollama/ollama:latest
    ports: ["11434:11434"]
    deploy:
      resources:
        reservations:
          devices:
            - capabilities: [gpu]
    volumes: ["ollama_data:/root/.ollama"]
  
  langfuse:
    image: langfuse/langfuse:2
    ports: ["3000:3000"]
    environment:
      - DATABASE_URL=postgresql://...
      - NEXTAUTH_SECRET=...
  
  nginx:
    image: nginx:alpine
    ports: ["80:80", "443:443"]
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./certs:/etc/nginx/certs
    depends_on: [api]

volumes:
  redis_data:
  chroma_data:
  ollama_data:

9 Conexão com RAG e próximos passos

Como MLOps completa o ciclo — do protótipo à produção sustentável.

O ciclo completo do RAG em produção

# Ciclo MLOps para RAG

# 1. Desenvolvimento (Módulos 0-10)
#    - Protótipo do RAG
#    - Fine-tuning do modelo
#    - Golden dataset criado

# 2. Avaliação (Módulo 11)
#    - RAGAS no golden dataset
#    - Métricas baseline estabelecidas
#    - Log no MLflow

# 3. Deploy (Módulo 11)
#    - Docker Compose / Kubernetes
#    - Langfuse para tracing
#    - Prometheus para métricas

# 4. Monitoramento (Módulo 11)
#    - Métricas em tempo real
#    - Detecção de drift
#    - Alertas automáticos

# 5. Melhoria contínua (Módulo 11)
#    - A/B testing de prompts
#    - Atualização do golden dataset
#    - Re-treinamento quando necessário

# 6. Responsabilidade (Módulo 12)
#    - Auditoria de viés
#    - Conformidade LGPD
#    - Documentação de impactos

O sistema final da Nimbus Cloud

Checklist final de produção

🎯 Quiz — teste seu conhecimento

Clique em uma alternativa para ver se acertou.

→ O que vem a seguir?

Agora que você sabe colocar IA em produção, vamos explorar a responsabilidade ética.

Conceitos que vamos construir aqui